package org.gol.paxos.member;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.gol.paxos.Message;
import org.gol.paxos.Proposal;
import org.gol.paxos.role.Acceptor;
import org.gol.paxos.role.MessageTranser;

/**
 * @package org.gol.paxos.client.member
 * @description 【参议员】 记录提案 承诺预备提案 批示提案 
 * @author Goliath
 * @createTime 2012-11-21 下午1:33:36
 * @modifyTime
 */
public class Senator implements Acceptor {
	private ExecutorService exec = Executors.newCachedThreadPool();

	private BlockingQueue<Message> messageQueue = new ArrayBlockingQueue<Message>(10);

	/**
	 * 议员都带了个文件夹，来记录提案 【内容和状态】
	 */
	public Map<String, Proposal> folder = Collections.synchronizedMap(new HashMap<String, Proposal>());
	/**
	* 参议员在议会的坐席号
	*/
	private String positionNo;

	/**
	 * 参议员的就位状态
	 */
	private boolean standby = false;

	public Senator(String positionNo) {
		this.setPositionNo(positionNo);
	}

	private Runnable receiveMessage = new Runnable() {
		public void run() {
			while (true) {
				try {
					Message oMessage = getMessageQueue().take();
					receive(oMessage);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
	};

	public String getPositionNo() {
		return positionNo;
	}

	public void setPositionNo(String positionNo) {
		this.positionNo = positionNo;
	}

	@Override
	public void receive(Message oMessage) {
		if (this.standby == false) {
			System.out.println("参议员不在岗位");
			return;
		}
		System.out.println(getPositionNo() + "开始处理来自" + oMessage.getFrom() + "的" + oMessage.getCommand() + "消息，消息列表里还有"
				+ getMessageQueue().size());
		String command = oMessage.getCommand();
		// 从文件袋里面撸出这个提案
		Proposal proposalMessage = (Proposal) oMessage.getContent();
		// 参议院拿出文件夹看看之前有木有人提出过这个提案
		Proposal proposalLocal = this.folder.get(proposalMessage.getContent());
		if ("prepare".equals(command)) {
			// 预处理操作
			if (proposalLocal == null) {
				// 若不存在提案 接收此提案 【一个批准者必须接受他收到的第一个提案】
				doAgree(oMessage.getFrom(), proposalMessage);
			} else {
				String proNoMessage = proposalMessage.getProNo();
				String proNoLocal = proposalLocal.getProNo();
				// 存在提案
				if (proNoMessage.compareTo(proNoLocal) < 0) {
					// 提案号小于本地提案的提案号
					// 直接打回去
					proposalMessage.setComment("参议员" + this.getPositionNo() + " 表示文件夹中的提案编号" + proNoLocal + " 大于 "
							+ proNoMessage);
					doDisagree(oMessage.getFrom(), proposalMessage);
				} else if (proNoMessage.compareTo(proNoLocal) > 0) {
					// 提案号大于本地提案的提案号
					// 校验提案内容 【一个值为V的提案被通过了，那么编号比这个提案大的提案的值应该是V】
					// 【提出一个编号为n具有值v的提案的前提是：存在一个多数派，要么他们中没有人批准过编号小于n的任何提案，要么他们批准的提案中编号小于n的最大的提案值是v】
					if (proposalMessage.getResult() == proposalLocal.getResult()) {
						doAgree(oMessage.getFrom(), proposalMessage);
					} else {
						proposalMessage.setComment("参议员" + this.getPositionNo() + " 表示文件夹中的提案编号" + proNoLocal + "值已为 "
								+ proposalLocal.getResult());
						doDisagree(oMessage.getFrom(), proposalMessage);
					}
				} else {
					// 提案号等于本地提案的提案号
					doAgree(oMessage.getFrom(), proposalMessage);
				}
			}
		} else if ("confirm".equals(command)) {
			// 标注为提交状态的只接受状态
			this.folder.put(proposalMessage.getContent(), proposalMessage);
		}
	}

	private void doDisagree(String from, Proposal proposalMessage) {
		proposalMessage.setState(Proposal.DISAGREE);
		proposalMessage.setAcceptorName(this.getPositionNo());
		sendCommand("disagree", from, proposalMessage);
	}

	private void doAgree(String from, Proposal proposalMessage) {
		// 签署Agree
		proposalMessage.setState(Proposal.AGREE);
		proposalMessage.setAcceptorName(this.getPositionNo());
		// 把这份提案复印了一份放到参议员自己的文件夹里
		try {
			this.folder.put(proposalMessage.getContent(), (Proposal) proposalMessage.clone());
		} catch (CloneNotSupportedException e) {
			e.printStackTrace();
		}
		sendCommand("agree", from, proposalMessage);
	}

	private void sendCommand(String command, String from, Proposal proposalMessage) {
		Message oMessage = new Message();
		oMessage.setCommand(command);
		oMessage.setFrom(this.getPositionNo());
		oMessage.setSendTo(from);
		oMessage.setContent(proposalMessage);
		MessageTranser oMessageTranser = new Servent();
		oMessageTranser.transMessage(oMessage);
	}

	public void standBy() {
		if (this.standby == false) {
			exec.submit(this.receiveMessage);
			System.out.println("参议员" + this.getPositionNo() + "就位");
			this.standby = true;
		}
	}

	public BlockingQueue<Message> getMessageQueue() {
		return messageQueue;
	}

}
